package br.ufc.great.dtx.tm.sch;

import java.io.IOException;
import java.io.ObjectOutputStream;
import java.net.Socket;
import java.util.Vector;

import br.ufc.great.Graph;
import br.ufc.great.dataguide.DGElement;
import br.ufc.great.dtx.server.DTXOutput;
import br.ufc.great.dtx.tm.Transaction;
import br.ufc.great.dtx.tm.lm.LockManager;

/**
 * 
 * @author Leonardo Oliveira Moreira
 *
 * Classe que implementa uma parte do escalonador que processa as mensagens remotas
 */
public class SchedulerRemoteMessage extends Thread {

	private DTXOutput output;
	private Scheduler scheduler;
	private LockManager lockManager;
	private SchedulerRemoteOperation schedulerRemoteOperation;
	
	// Lista de messagens remotas (commits e aborts)
	private Vector<RemoteMessage> remoteMessages;
	
	public SchedulerRemoteMessage(DTXOutput output, Scheduler scheduler, LockManager lockManager, SchedulerRemoteOperation schedulerRemoteOperation) {
		this.output = output;
		this.scheduler = scheduler;
		this.lockManager = lockManager;
		this.schedulerRemoteOperation = schedulerRemoteOperation;
		remoteMessages = new Vector<RemoteMessage>();
		setPriority(MIN_PRIORITY);
	}
	
	public void run() {
		while (isAlive()) {
			// Processa messagens remotas
			while (remoteMessages.size() > 0) {
				RemoteMessage remoteMessage = remoteMessages.remove(0);
				//output.print(getClass().getSimpleName(), "processing message", DTXOutput.TEST);
				switch (remoteMessage.getMessage().getType()) {
					case Message.DISTRIBUTED_ABORT : {
						//output.print(getClass().getSimpleName(), "Processing distributed abort remote message...", DTXOutput.REMOTE_MESSAGE);
						scheduler.abort(remoteMessage.getMessage().getTransactionId());
						break;
					}
					case Message.ABORT : {
						processAbortRemoteMessage(remoteMessage);
						break;
					}
					case Message.COMMIT : {
						processCommitRemoteMessage(remoteMessage);
						break;
					}
				}
				//output.print(getClass().getSimpleName(), "message has processed", DTXOutput.TEST);
			}
			if (remoteMessages.size() > 0)
				continue;
			pause();

		}
	}
	
	public synchronized void pause() {
		try {
			output.print(getClass().getSimpleName(), "is waiting", DTXOutput.TEST);
			wait();
		}
		catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
	
	public synchronized void restart() {
		notify();
	}
	
	public synchronized void addRemoteMessage(RemoteMessage remoteMessage) {
		Message message = remoteMessage.getMessage();
		output.print(getClass().getSimpleName(), "Remote message of transaction " + message.getTransactionId() + " was received", DTXOutput.TEST);
		remoteMessages.add(remoteMessage);
		restart();
	}

	private synchronized void processAbortRemoteMessage(RemoteMessage remoteMessage) {
		output.print(getClass().getSimpleName(), "Processing abort remote message...", DTXOutput.TEST);
		synchronized (lockManager) {
			lockManager.unlock(remoteMessage.getMessage().getTransactionId());
		}
		Graph waitFor = scheduler.getWaitForGraph();
		synchronized (waitFor) {
			waitFor.removerVertex(remoteMessage.getMessage().getTransactionId());			
		}
		remoteMessage.getMessage().setStatus(Message.EXECUTED);
		Socket socket = remoteMessage.getSocket();
		try {
			ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream());
			out.writeObject(remoteMessage.getMessage());
			out.flush();
		} catch (IOException e) {
			e.printStackTrace();
			output.print(getClass().getSimpleName(), e.getMessage(), DTXOutput.ERROR);
		}
		output.print(getClass().getSimpleName(), "Abort remote message executed", DTXOutput.TEST);
	}
	
	private synchronized void processCommitRemoteMessage(RemoteMessage remoteMessage) {
		output.print(getClass().getSimpleName(), "Processing commit remote message...", DTXOutput.TEST);
		synchronized (lockManager) {
			lockManager.unlock(remoteMessage.getMessage().getTransactionId());
		}
		Graph waitFor = scheduler.getWaitForGraph();
		synchronized (waitFor) {
			waitFor.removerVertex(remoteMessage.getMessage().getTransactionId());			
		}
		remoteMessage.getMessage().setStatus(Message.EXECUTED);
		Socket socket = remoteMessage.getSocket();
		try {
			ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream());
			out.writeObject(remoteMessage.getMessage());
			out.flush();
		} catch (IOException e) {
			e.printStackTrace();
			output.print(getClass().getSimpleName(), e.getMessage(), DTXOutput.ERROR);
		}
		output.print(getClass().getSimpleName(), "Commit remote message executed", DTXOutput.TEST);
	}

	public synchronized int size() {
		return remoteMessages.size();
	}

	/*
	public void removeRemoteMessages(String transactionId) {
		synchronized (remoteMessages) {
			for (int i = 0; i < remoteMessages.size(); i++) {
				RemoteMessage remoteMessage = remoteMessages.elementAt(i);
				if (remoteMessage.getMessage().getTransactionId().equalsIgnoreCase(transactionId)) {
					try {
						remoteMessage.getSocket().close();
					} catch (IOException e) {

					}
					remoteMessages.remove(i);
					i--;
				}
			}
		}
	}
	*/
	
}
